package avicit.bdp.dds.server.master.runner;

import avicit.bdp.common.utils.redis.RedisCacheHelper;
import avicit.bdp.dds.api.service.ProcessResourceConfigService;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.common.thread.Stopper;
import avicit.bdp.dds.common.thread.ThreadUtils;
import avicit.bdp.dds.common.utils.OSUtils;
import avicit.bdp.dds.dao.entity.Command;
import avicit.bdp.dds.dao.entity.ProcessInstance;
import avicit.bdp.dds.remote.NettyRemotingClient;
import avicit.bdp.dds.remote.config.NettyClientConfig;
import avicit.bdp.dds.server.master.config.MasterConfig;
import avicit.bdp.dds.server.zk.ZKMasterClient;
import avicit.bdp.dds.service.process.ProcessService;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * master scheduler thread
 */
@Service
public class MasterSchedulerService extends Thread {

    /**
     * logger of MasterSchedulerThread
     */
    private static final Logger logger = LoggerFactory.getLogger(MasterSchedulerService.class);

    /**
     * dolphinscheduler database interface
     */
    @Autowired
    private ProcessService processService;

    /**
     * zookeeper master client
     */
    @Autowired
    private ZKMasterClient zkMasterClient;

    /**
     * master config
     */
    @Autowired
    private MasterConfig masterConfig;

    @Autowired
    private ProcessResourceConfigService processResourceConfigService;

    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * netty remoting client
     */
    private NettyRemotingClient nettyRemotingClient;

    /**
     * master exec service
     */
    private ThreadPoolExecutor masterExecService;


    /**
     * constructor of MasterSchedulerThread
     */
    @PostConstruct
    public void init() {
        this.masterExecService = (ThreadPoolExecutor) ThreadUtils.newDaemonFixedThreadExecutor("Master-Exec-Thread", masterConfig.getMasterExecThreads());
        NettyClientConfig clientConfig = new NettyClientConfig();
        this.nettyRemotingClient = new NettyRemotingClient(clientConfig);
    }

    @Override
    public void start() {
        super.setName("MasterSchedulerThread");
        super.start();
    }

    public void close() {
        masterExecService.shutdown();
        boolean terminated = false;
        try {
            terminated = masterExecService.awaitTermination(5, TimeUnit.SECONDS);
        } catch (InterruptedException ignore) {
        }
        if (!terminated) {
            logger.warn("masterExecService shutdown without terminated, increase await time");
        }
        nettyRemotingClient.close();
        logger.info("master schedule service stopped...");
    }

    /**
     * run of MasterSchedulerThread
     */
    @Override
    public void run() {
        logger.info("master scheduler started");
        while (Stopper.isRunning()) {
            InterProcessMutex mutex = null;
            try {
                boolean runCheckFlag = OSUtils.checkResource(masterConfig.getMasterMaxCpuloadAvg(), masterConfig.getMasterReservedMemory());
                if (!runCheckFlag) {
                    Thread.sleep(Constants.SLEEP_TIME_MILLIS);
                    continue;
                }
                if (zkMasterClient.getZkClient().getState() == CuratorFrameworkState.STARTED) {

                    mutex = zkMasterClient.blockAcquireMutex();

                    int activeCount = masterExecService.getActiveCount();
                    // make sure to scan and delete command  table in one transaction
                    Command command = processService.findOneCommand();
                    if (command != null) {
                        logger.info("find one command: id: {}, type: {}", command.getId(), command.getCommandType());

                        try {

                            ProcessInstance processInstance = processService.handleCommand(logger,
                                    getLocalAddress(),
                                    this.masterConfig.getMasterExecThreads() - activeCount, command);
                            if (processInstance != null) {
                                //把启动的流程实例id放入到redis缓存中，方便启动程序获取
                                RedisCacheHelper.getInstance().set("bdp:dds:processDefinitionId:" + processInstance.getProcessDefinitionId(), processInstance.getId());
                                RedisCacheHelper.getInstance().expire("bdp:dds:task:ProcessDefinitionId:" + processInstance.getProcessDefinitionId(), 120);
                                logger.info("实例ID：" + RedisCacheHelper.getInstance().get("bdp:dds:processDefinitionId:" + processInstance.getProcessDefinitionId()));
                                logger.info("start master exec thread , split DAG ..." + processInstance.getProcessDefinitionId() + "--" + processInstance.getId());
                                masterExecService.execute(new MasterExecThread(processInstance, processService, processResourceConfigService, nettyRemotingClient, redisTemplate));
                            }
                        } catch (Exception e) {
                            logger.error("scan command error ", e);
                            processService.moveToErrorCommand(command, e.toString());
                        }
                    } else {
                        //indicate that no command ,sleep for 1s
                        Thread.sleep(Constants.SLEEP_TIME_MILLIS);
                    }
                }
            } catch (Exception e) {
                logger.error("master scheduler thread error", e);
            } finally {
                zkMasterClient.releaseMutex(mutex);
            }
        }
    }

    private String getLocalAddress() {
        return OSUtils.getHost() + ":" + masterConfig.getListenPort();
    }
}
